package hou.report

import java.util.Properties

import com.google.gson.Gson
import hou.beans.ProvinceCityResult
import hou.config.ConfigHelper
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import scalikejdbc.{DB, SQL}



/**
  * 统计各省市分布情况sparkCore
  */
object ProCityAnalysisCore {
  def main(args: Array[String]): Unit = {
    //sparkContext
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getName}")
    conf.setMaster("local[*]")
    conf.set("spark.serializer",ConfigHelper.serializer)

    val sc = new SparkContext(conf)
    val sQLContext = new SQLContext(sc)

    //读取数据parquet
    val dataframe = sQLContext.read.parquet(ConfigHelper.parquetPath)
    //core reduceByKey -> RDD[k(省市),1]

    val resultRDD: RDD[((String, String), Int)] = dataframe.map(row => {
      val pname = row.getAs[String]("provincename")
      val cname = row.getAs[String]("cityname")
      ((pname, cname), 1)
    }).reduceByKey(_ + _)
    resultRDD

    //JSON  第一种方式
    resultRDD.map(tp=>{
      val gson = new Gson()
      gson.toJson(ProvinceCityResult(tp._1._1,tp._1._2,tp._2))
    }).coalesce(1).saveAsTextFile("C:\\Users\\44323\\Desktop\\资料PDF\\111")

    //json 第二种方式
    import sQLContext.implicits._
    val datafram = resultRDD.map(tp=>(tp._1._1,tp._1._2,tp._2)).toDF("pname","cname","count")
    datafram.coalesce(1).write.json("C:\\Users\\44323\\Desktop\\资料PDF\\222")

    //写入mysql  第一种方式

    val props = new Properties()
    props.setProperty("driver",ConfigHelper.driver)
    props.setProperty("user",ConfigHelper.user)
    props.setProperty("password",ConfigHelper.password)
    datafram.write.jdbc(ConfigHelper.url,ConfigHelper.table,props)

    //写入mysql  第二种方式
    resultRDD.foreachPartition(partition=>{
      DB.localTx(implicit session=>{
        partition.foreach(tp=>{
          SQL("insert into ProCity_33 (pname,cname,count) values (?,?,?)")
            .bind(tp._1._1,tp._1._2,tp._2).update().apply()
        })
      })
    })



    sc.stop()
  }
}
